package com.rocketmq.one.message;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.Date;
import java.util.List;

public class RetryMessage {
    @Test
    public void retryMessage() throws Exception {
        // 获取生产者对象
        DefaultMQProducer producer = new DefaultMQProducer("retry-produce-group");
        // 设置服务对象
        producer.setNamesrvAddr("8.134.163.217:9876");
        // 开始
        producer.start();
        // 设置重试次数
        producer.setRetryTimesWhenSendFailed(3);
        Message message = new Message("retryTopic", "我是vip666的文章".getBytes());
        producer.send(message);
        System.out.println("发送成功");
        producer.shutdown();
    }
    /**
     * 重试的时间间隔
     * 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * 默认重试16次
     * 1.能否自定义重试次数
     * 2.如果重试了16次(并发模式) 顺序模式下(int最大值次)都是失败的?  是一个死信消息 则会放在一个死信主题中去 主题的名称:%DLQ%retry-consumer-group
     * 3.当消息处理失败的时候 该如何正确的处理?
     * --------------
     * 重试的次数一般 5次
     * @throws Exception
     */
    @Test
    public void retryConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");
        consumer.setNamesrvAddr("8.134.163.217:9876");
        consumer.subscribe("retryTopic", "*");
        // 设定重试次数
        consumer.setMaxReconsumeTimes(2);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = msgs.get(0);
                System.out.println(new Date());
                System.out.println(messageExt.getReconsumeTimes());
                System.out.println(new String(messageExt.getBody()));
                // 业务报错了 返回null 返回 RECONSUME_LATER 都会重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
        System.in.read();
    }





}
